#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__
#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "../mqcommon/mq_proto.pb.h"
#include "mq_consumer.hpp"
#include "mq_route.hpp"
#include "mq_host.hpp"

namespace mymq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string& cid,
                const VirtualHost::ptr& host,
                const ConsumerManager::ptr& cmp,
                const ProtobufCodecPtr& codec,
                const muduo::net::TcpConnectionPtr& conn,
                const ThreadPool::ptr& pool)
        : _cid(cid),
          _conn(conn),
          _codec(codec),
          _cmp(cmp),
          _host(host),
          _pool(pool)
        {}

        ~Channel()
        {
            // 信道销毁，那么所关联的消费者也要销毁
            if (_consumer.get() != nullptr)
                _cmp->remove(_consumer->tag, _consumer->qname);
        }

        // 交换机的声明与删除
        void declareExchange(const declareExchangeRequestPtr& req)
        {
            bool ret = _host->declareExchange(req->exchange_name(), req->type(), 
                req->durable(), req->auto_delete(), req->args());
            // 构造一个通用响应交给 codec 进行协议处理并发送
            basicResponse(ret, req->rid(), req->cid());
        }
        void deleteExchange(const deleteExchangeRequestPtr& req)
        {
            _host->deleteExchange(req->exchange_name());
            basicResponse(true, req->rid(), req->cid());
        }

        // 队列的声明与删除
        void declareQueue(const declareQueueRequestPtr& req)
        {
            bool ret = _host->declareQueue(req->queue_name(), req->durable(), req->exclusive(),
                req->auto_delete(), req->args());
            // 如果队列声明失败，就不需要初始化 队列消费者句柄 了
            if (ret == false)
                return basicResponse(ret, req->rid(), req->cid());
            // 初始化 队列消费者 结构
            _cmp->initQueueConsumer(req->queue_name());
            basicResponse(ret, req->rid(), req->cid());
        }
        void deleteQueue(const deleteQueueRequestPtr& req)
        {
            // 删除队列，同时删除 队列消费者 管理句柄
            _cmp->destroyQueueConsumer(req->queue_name());
            _host->deleteQueue(req->queue_name());
            basicResponse(true, req->rid(), req->cid());
        }

        // 队列的绑定与解绑
        void bind(const queueBindRequestPtr& req)
        {
            int ret = _host->bind(req->exchange_name(), req->queue_name(), req->binding_key());
            basicResponse(ret, req->rid(), req->cid());
        }
        void unBind(const queueUnBindRequestPtr& req)
        {
            _host->unBind(req->exchange_name(), req->queue_name());
            basicResponse(false, req->rid(), req->cid());
        }

        // 消息发布，channel角色变为了生产者
        void basicPublish(const basicPublishRequestPtr& req)
        {
            DLOG("开始进行消息发布");
            // 消息发布到交换机，由交换机进行交换路由，把消息发送到符合条件的队列中
            // 再将 消息消费 任务交给线程池，完成消息推送到客户端
            // 1.获取指定交换机
            Exchange::ptr ep = _host->selectExchange(req->exchange_name());
            if (ep.get() == nullptr)
                return basicResponse(false, req->rid(), req->cid());

            // 获取交换机绑定的队列
            MsgQueueBindingMap queue_binding_map = _host->getExchangeBindings(req->exchange_name());
            // 2.进行交换路由，将消息发布到队列中
            BasicProperties* properties = nullptr;
            std::string routing_key;
            if (req->has_properties())
            {
                // 获取消息的属性和routingkey
                properties = req->mutable_properties();
                routing_key = properties->routing_key();
            }
            // route
            for (auto& binding : queue_binding_map)
            {
                if (Router::Route(ep->type, routing_key, binding.second->binding_key))
                {
                    // 3.路由匹配成功，将消息添加到队列中
                    _host->basicPublish(binding.first, properties, req->body());
                    // 4.向线程池中添加 消息消费 任务：从消息队列中取出一条消息，取出一个消费者进行消费
                    auto task = std::bind(&Channel::consume, this, binding.first);
                    // DLOG("线程池：%p", _pool.get());
                    _pool->push(task);
                }
            }
            return basicResponse(true, req->rid(), req->cid());
        }
        // 消息确认
        void basicAck(const basicAckRequestPtr& req)
        {
            _host->basicAck(req->queue_name(), req->msg_id());
            return basicResponse(true, req->rid(), req->cid());
        }

        // 订阅队列消息
        void basicConsume(const basicConsumeRequestPtr& req)
        {
            // 订阅消息后，channel的角色就变成了 消费者
            // 判断队列是否存在
            bool ret = _host->existsQueue(req->queue_name());
            if (ret == false)
                return basicResponse(true, req->rid(), req->cid());

            // 队列存在，创建消费者
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1, std::placeholders::_2,
                std::placeholders::_3);
            _consumer = _cmp->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
            return basicResponse(true, req->rid(), req->cid());
        }
        // 取消订阅
        void basicCancel(const basicCancelRequestPtr& req)
        {
            _cmp->remove(req->consumer_tag(), req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }
    private:
        // 消费者回调函数，将消息推送给客户端
        void callback(const std::string& ctag, const BasicProperties* bp, const std::string& body)
        {
            // 构造消息应答
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_consumer_tag(ctag);
            resp.set_body(body);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            // 交给协议处理句柄 codec 来处理协议，并发送
            _codec->send(_conn, resp);
        }
        // 消息消费，线程池完成
        void consume(const std::string& qname)
        {
            // 1.获取一条消息
            MessagePtr msg = _host->basicConsume(qname);
            if (msg.get() == nullptr)
            {
                DLOG("执行消费任务失败：%s 队列没有消息", qname.c_str());
                return;
            }
            // 2.获取一个订阅者
            Consumer::ptr cp = _cmp->choose(qname);
            if (cp.get() == nullptr)
            {
                DLOG("执行消费任务失败：%s 队列没有消费者", qname.c_str());
                return;
            }
            // 3.调用订阅者的回调函数进行消费（将消息推送给客户端）
            cp->callback(cp->tag, msg->mutable_payload()->mutable_properties(), msg->payload().body());
            // 4.判断订阅者是否是自动确认
            if (cp->auto_ack)
                _host->basicAck(qname, msg->payload().properties().id());
        }

        // 发送通用的响应
        void basicResponse(bool ret, const std::string& rid, const std::string& cid)
        {
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ret);
            _codec->send(_conn, resp);
        }
        
    private:
        std::string _cid;
        Consumer::ptr _consumer; // 信道关联的消费者
        muduo::net::TcpConnectionPtr _conn; // 信道关联的连接
        ProtobufCodecPtr _codec; // proto 协议处理
        ConsumerManager::ptr _cmp; // 消费者管理句柄
        VirtualHost::ptr _host; // 虚拟机句柄
        ThreadPool::ptr _pool; // 线程池
    };
    
    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        bool openChannel(const std::string& cid,
            const VirtualHost::ptr& host,
            const ConsumerManager::ptr& cmp,
            const ProtobufCodecPtr& codec,
            const muduo::net::TcpConnectionPtr& conn,
            const ThreadPool::ptr& pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 查看是否重复
            auto it = _channels.find(cid);
            if (it != _channels.end())
            {
                DLOG("信道 %s 已存在", cid.c_str());
                return false;
            }
            // 构造 channel
            Channel::ptr channel = std::make_shared<Channel>(cid, host, cmp, codec, conn, pool);
            _channels.insert(std::make_pair(cid, channel));
            return true;
        }

        void closeChannel(const std::string& id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);            
        }
        Channel::ptr getChannel(const std::string& id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 查看是否存在
            auto it = _channels.find(id);
            if (it == _channels.end())
                return Channel::ptr();

            return it->second;
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}


#endif